package org.redis.cluster.transport.netty.client;

import java.util.concurrent.TimeUnit;

import org.redis.cluster.Configs;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.transport.client.Client;
import org.redis.cluster.transport.exception.TimeoutException;
import org.redis.cluster.transport.exception.TransportException;
import org.redis.cluster.transport.netty.protocol.NettyDecoder;
import org.redis.cluster.transport.netty.protocol.NettyEncoder;
import org.redis.cluster.transport.netty.session.NettySessionContext;
import org.redis.cluster.transport.protocol.ProtocolDecoder;
import org.redis.cluster.transport.protocol.ProtocolEncoder;
import org.redis.cluster.transport.session.SessionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient implements Client {
	
	private final static EventLoopGroup CLIENT_LOOP_GROUP=new NioEventLoopGroup(Configs.DEFAULT_SERVER_THREAD);
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private Channel channel;
	private NettyClientHandler handler;
	
	private ClientConfig config;
	private ProtocolDecoder decoder;
	private ProtocolEncoder encoder;
	
	public NettyClient(ClientConfig config,ProtocolDecoder decoder,ProtocolEncoder encoder) {
		this.config=config;
		this.decoder=decoder;
		this.encoder=encoder;
		this.handler=initHandler();
		this.channel=initChannel();
	}
	
	protected Channel initChannel() {
		Bootstrap bootstrap=new Bootstrap();
		
		bootstrap.group(CLIENT_LOOP_GROUP);
		bootstrap.channel(NioSocketChannel.class);
		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel socketChannel) throws Exception {
            	SessionContext context=new NettySessionContext();
            	ChannelPipeline pipeline=socketChannel.pipeline();
            	pipeline.addLast("decoder", getDecoder(context));
            	pipeline.addLast("encoder", getEncoder(context));
            	pipeline.addLast("handler", getHandler(context));
            }
        });

		try {
			ChannelFuture future=bootstrap.connect(config.getHost(), config.getPort()).sync();
			return future.channel();
		}
		catch(Exception e) {
			throw new TransportException(e);
		}
	}
	
	protected NettyClientHandler initHandler() {
		return new NettyClientHandler();
	}
	
	protected ChannelHandler getHandler(SessionContext context) {
		return handler;
	}

	protected ChannelHandler getEncoder(SessionContext context) {
		return new NettyEncoder(encoder);
	}

	protected ChannelHandler getDecoder(SessionContext context) {
		return new NettyDecoder(decoder,context);
	}

	@Override
	public byte[] send(Object data) throws TimeoutException {
		try {
			channel.writeAndFlush(data);
			byte response[]=handler.getResponse(config.getTimeout(),TimeUnit.MILLISECONDS);
			if(response==null) {
				throw new TimeoutException();
			}
			
			return response;
		}
		catch(InterruptedException e) {
			if(logger.isErrorEnabled()) {
				logger.error("",e);
			}
			
			throw new TimeoutException();
		}
	}
	
	@Override
	public void close() {
		try {
			channel.close().sync();
		}
		catch(Exception e) {
			throw new TransportException(e);
		}
	}
}